package com.atguigu.tingshu.search.receiver;

import com.atguigu.tingshu.common.rabbit.constant.MqConst;
import com.atguigu.tingshu.search.service.SearchService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

/**
 * @BelongsProject: tingshu-parent
 * @BelongsPackage: com.atguigu.tingshu.search.receiver
 * @Author: jiangyi
 * @CreateTime: 2025-10-09  17:06
 * @Description: TODO
 * @Version: 1.0
 */
@Component
@Slf4j
public class AlbumInfoIndexListener {

    @Autowired
    private SearchService searchService;


    /**
     * 上架专辑至 ES
     *
     * @param albumId
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = MqConst.EXCHANGE_ALBUM, durable = "true"),
            value = @Queue(value = MqConst.QUEUE_ALBUM_UPPER, durable = "true"),
            key = {MqConst.ROUTING_ALBUM_UPPER}
    ))
    public void upperAlbumInfoIndex(Long albumId, Message message, Channel channel) {
        // 验证channel是否有效
        if (channel == null || !channel.isOpen()) {
            log.error("Channel is closed or null for albumId: {}", albumId);
            return;
        }
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            searchService.upperAlbumInfoIndex(albumId);

            // 确认消息前再次检查channel状态
            if (channel.isOpen()) {
                channel.basicAck(deliveryTag, false);
                log.info("上架专辑成功：{}", albumId);
            } else {
                log.error("Channel closed before ack for albumId: {}", albumId);
            }

        } catch (Exception e) {
            log.error("更新专辑索引失败，albumId: {}", albumId, e);

            try {
                // 拒绝消息前检查channel状态
                if (channel.isOpen()) {
                    channel.basicNack(deliveryTag, false, true);
                    log.info("消息已拒绝并重新入队，albumId: {}", albumId);
                }
            } catch (IOException ioEx) {
                log.error("消息拒绝失败，albumId: {}", albumId, ioEx);
            }

            // 注意：这里不要重新抛出异常，避免循环重试
            // throw new RuntimeException("处理消息失败", e);
        }
    }

    /**
     * ES 中下架专辑
     *
     * @param albumId
     * @param message
     * @param channel
     */
    @RabbitListener(bindings = @QueueBinding(
            exchange = @Exchange(value = MqConst.EXCHANGE_ALBUM, durable = "true"),
            value = @Queue(value = MqConst.QUEUE_ALBUM_LOWER, durable = "true"),
            key = {MqConst.ROUTING_ALBUM_LOWER}
    ))
    public void lowerAlbumInfoIndex(Long albumId, Message message, Channel channel) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {

            searchService.lowerAlbumInfoIndex(albumId);
            channel.basicAck(deliveryTag, false);
            log.info("下架专辑：{}", albumId);
        } catch (Exception e) {
            try {
                channel.basicNack(deliveryTag, false, true);
            } catch (IOException ex) {
                log.error("消息拒绝失败，albumId: {}", albumId, ex);
            }
            log.error("下架专辑索引失败，albumId: {}", albumId, e);
            throw new RuntimeException("处理消息失败", e);
        }

    }


}
